package org.baojie.raft.connector.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.baojie.raft.box.Box;
import org.baojie.raft.connector.client.initializer.ClientChannelInit;
import org.baojie.raft.node.NodeAddress;
import org.baojie.raft.node.NodeName;
import org.baojie.raft.connector.ChannelKey;
import org.baojie.raft.util.current.RaftFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;

public class RaftNettyClient {

    private static final Logger log = LoggerFactory.getLogger(RaftNettyClient.class);

    private final ConcurrentHashMap<ChannelKey, Channel> channels = new ConcurrentHashMap<>(512);

    private final boolean epollAvailable;
    private final EventLoopGroup eventLoop;
    private final int works;
    private final Box box;

    public RaftNettyClient(int workNums, Box box) {
        this.box = box;
        this.works = workNums;
        this.epollAvailable = Epoll.isAvailable();
        if (epollAvailable) {
            RaftFactory factory = RaftFactory.create("epoll raft netty client worker", false);
            eventLoop = new EpollEventLoopGroup(works, factory);
        } else {
            RaftFactory factory = RaftFactory.create("nio raft netty client worker", false);
            eventLoop = new NioEventLoopGroup(works, factory);
        }
    }

    public void stop() {
        try {
            if (null != eventLoop) {
                eventLoop.shutdownGracefully();
            }
        } finally {
            Iterator<Channel> iterator = channels.values().iterator();
            while (iterator.hasNext()) {
                Channel tmp = iterator.next();
                tmp.close();
            }
            channels.clear();
        }
    }

    public Channel getChannel(NodeName remote) {
        if (null == remote) {
            return null;
        }
        final ChannelKey key = new ChannelKey(remote);
        Channel channel = channels.get(key);
        if (null != channel) {
            return channel;
        } else {
            return createNewer(key, remote);
        }
    }

    private Channel createNewer(ChannelKey key, NodeName remote) {
        Bootstrap bootstrap = new Bootstrap().group(eventLoop);
        if (epollAvailable) {
            bootstrap.channel(EpollSocketChannel.class);
        } else {
            bootstrap.channel(NioSocketChannel.class);
        }

        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30000)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ClientChannelInit(box));
        NodeAddress address = remote.getAddress();

        ChannelFuture future = bootstrap.connect(address.getHost(), address.getPort());
        future.awaitUninterruptibly();

        if (future.isCancelled()) {
            return null;
        } else if (!future.isSuccess()) {
            return null;
        } else {
            Channel channel = future.channel();
            if (null == channels.putIfAbsent(key, channel)) {
                channel.closeFuture().addListener((ChannelFutureListener) cf -> {
                    channels.remove(key);
                    log.warn("");
                });
                return channel;
            } else {
                channel.closeFuture().addListener((ChannelFutureListener) cf -> {
                    log.warn("");
                });
                channel.close();
                return channels.get(key);
            }
        }
    }

}
